package com.mooc.kafka.stream;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;

import java.util.Arrays;
import java.util.Locale;
import java.util.Properties;

/**
 * @author StarsOfFuture_xYang
 * @version 1.0
 * @date 2021-04-12 10:31 下午
 * @information kafka-study - com.mooc.kafka.stream
 **/
public class StreamSample {

    public static final String INPUT_TOPIC = "jiangzh-stream-in";
    public static final String OUT_TOPIC = "jiangzh-stream-out";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "1.116.23.168:9092");
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordCount-app");
        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        //构建流拓扑结构
        final StreamsBuilder builder = new StreamsBuilder();
//        wordCount(builder);
        //构建foreachStream
        wordCount1(builder);
        KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties);
        kafkaStreams.start();
    }

    //如果定义流计算过程
    static void wordCount1(final StreamsBuilder streamsBuilder) {
        KStream<String, String> source = streamsBuilder.stream(INPUT_TOPIC);
        source.flatMapValues(
                value -> Arrays.asList(
                        value.toLowerCase(
                                Locale.getDefault()).split(" "))).
                foreach((key,value)-> System.out.println(key+" : "+value));
    }

    //如果定义流计算过程
    static void wordCount(final StreamsBuilder streamsBuilder) {
        KStream<String, String> source = streamsBuilder.stream(INPUT_TOPIC);
        KTable<String, Long> count = source.flatMapValues(
                value -> Arrays.asList(
                        value.toLowerCase(
                                Locale.getDefault()).split(" "))).
                groupBy((key, value) -> value).count();
        count.toStream().to(OUT_TOPIC, Produced.with(Serdes.String(), Serdes.Long()));
    }
}
